spb/worthdoing Public

1/**2 * WorthDoing.ai3 * Author: Simon-Pierre Boucher4 * Contact: contact@spboucher.ai5 * File: src/app/api/investigations/[id]/events/route.ts6 * Description: SSE stream of real AgentEvents with Last-Event-ID replay; proxy-safe (no buffering, per-event flush).7 */8import { NextRequest } from "next/server";9import { z } from "zod";10import { subscribe, eventsAfter, type AgentEvent } from "@/lib/agent/events";1112export const dynamic = "force-dynamic";1314const HEARTBEAT_MS = 15_000;1516export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string }> }) {17  const { id } = await ctx.params;18  if (!z.string().uuid().safeParse(id).success) {19    return new Response("Invalid id", { status: 400 });20  }2122  const lastEventIdHeader = req.headers.get("last-event-id") ?? req.nextUrl.searchParams.get("lastEventId");23  const afterSeq = lastEventIdHeader ? parseInt(lastEventIdHeader, 10) || 0 : 0;2425  const encoder = new TextEncoder();2627  const stream = new ReadableStream({28    async start(controller) {29      let closed = false;30      const send = (event: AgentEvent) => {31        if (closed) return;32        try {33          controller.enqueue(34            encoder.encode(`id: ${event.seq}\nevent: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`),35          );36        } catch {37          closed = true;38        }39      };4041      // Live subscription first so nothing falls in the gap, then replay history.42      const seen = new Set<number>();43      const buffer: AgentEvent[] = [];44      let replaying = true;45      const unsubscribe = subscribe(id, (e) => {46        if (replaying) buffer.push(e);47        else if (!seen.has(e.seq)) {48          seen.add(e.seq);49          send(e);50        }51      });5253      const history = await eventsAfter(id, afterSeq);54      for (const e of history) {55        seen.add(e.seq);56        send(e);57      }58      replaying = false;59      for (const e of buffer) {60        if (!seen.has(e.seq)) {61          seen.add(e.seq);62          send(e);63        }64      }6566      const heartbeat = setInterval(() => {67        if (closed) return;68        try {69          controller.enqueue(encoder.encode(`: heartbeat ${Date.now()}\n\n`));70        } catch {71          closed = true;72        }73      }, HEARTBEAT_MS);7475      const cleanup = () => {76        closed = true;77        clearInterval(heartbeat);78        unsubscribe();79        try {80          controller.close();81        } catch {82          // already closed83        }84      };85      req.signal.addEventListener("abort", cleanup);86    },87  });8889  return new Response(stream, {90    headers: {91      "Content-Type": "text/event-stream; charset=utf-8",92      "Cache-Control": "no-cache, no-transform",93      Connection: "keep-alive",94      "X-Accel-Buffering": "no",95    },96  });97}98